package com.wsjj.gmall.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.wsjj.gmall.bean.TableProcessDim;
import com.wsjj.gmall.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.*;
import java.util.function.Predicate;

/**
 * 改类是广播后判断主流是否是维度流数据
 *
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {

    private MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;


    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }


    //        问题：
//              主流和广播流来的顺序无法控制，如果主流先到，广播流没有到，这样主流的关联不上会丢数据，解决办法：
//                  1.先启动程序，让配置表先跑，在启动maxwell，这样会影响到离线数仓，如果是  保存点或者检查点恢复   这样无法控制maxwell，恢复是kafka重放，所以和maxwell没有关系
//                  2.使用open（）方法提交预加载，然后存储到map集合中

    Map<String,TableProcessDim> hashmap  = new HashMap<String,TableProcessDim>();


        @Override
        public void open(Configuration parameters) throws Exception {

            java.sql.Connection mysqlConnection = JdbcUtil.createMysqlConnection();
            String sql = "select * from table_process_dim";
            List<TableProcessDim> tableProcessDims = JdbcUtil.selectList(mysqlConnection, sql, TableProcessDim.class, true);
            for (TableProcessDim tableProcessDim : tableProcessDims) {
                hashmap.put(tableProcessDim.getSourceTable(),tableProcessDim);
            }

            JdbcUtil.closeMysqlConnection(mysqlConnection);

        }

        @Override
        public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {


            String table = value.getString("table");

            ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
            TableProcessDim tableProcessDim = broadcastState.get(table);
            System.out.println("tableProcessDim========: "+tableProcessDim);

            if(tableProcessDim != null || hashmap.containsKey(table)){
//                    细节：1.向下传递的数据要插入或者从hbase中删除的，所以要携带  hbase表名和表信息    2.维度的变化是crud中的哪个也要往下传递
                System.out.println("维度表");
                String type = value.getString("type");

                JSONObject data = value.getJSONObject("data");
                System.out.println("jdaskldsjak"+type);
//                    细节：维度表中有些列不用要（不如：logo列），需要过滤掉

                deleNotNeedColumns(data,tableProcessDim);
                data.put("type",type);




                if(tableProcessDim != null){
                    out.collect(Tuple2.of(data,tableProcessDim));
                }else {
                    out.collect(Tuple2.of(data,hashmap.get(table)));
                }

            }








        }




        //            希望先执行这个方法
        @Override
        public void processBroadcastElement(TableProcessDim value, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {


            BroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

            String op = value.getOp();
            String sourceTable = value.getSourceTable();
            //              判断广播状态，如果是“d”就删除广播状态，不是就添加
            if("d".equals(op)){
                broadcastState.remove(sourceTable);
                hashmap.remove(sourceTable);
            }else {


                broadcastState.put(sourceTable,value);
                hashmap.put(sourceTable,value);


            }
            System.out.print(" dsajikliodjfkfidfk "+broadcastState.get(sourceTable));




        }



    public static void deleNotNeedColumns(JSONObject data, TableProcessDim tableProcessDim) {
        System.out.println("-------------------------------"+data+"----"+tableProcessDim);
        String sinkColumns = tableProcessDim.getSinkColumns();
        String[] sink = sinkColumns.split(",");
        List<String> list = Arrays.asList(sink);

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(new Predicate<Map.Entry<String, Object>>() {
            @Override
            public boolean test(Map.Entry<String, Object> stringObjectEntry) {
                return !list.contains(stringObjectEntry.getKey());
            }
        });



    }





}